package org.tigr.htc.request;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringReader;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.apache.xmlrpc.WebServer;
import org.apache.xmlrpc.XmlRpcClient;
import org.apache.xmlrpc.XmlRpcException;
import org.tigr.antware.shared.command.events.CommandInterrupt;
import org.tigr.antware.shared.command.events.CommandSuspend;
import org.tigr.antware.shared.command.events.FailureEvent;
import org.tigr.antware.shared.command.events.FinishEvent;
import org.tigr.antware.shared.command.events.InterruptEvent;
import org.tigr.antware.shared.command.events.ResumeEvent;
import org.tigr.antware.shared.command.events.StartEvent;
import org.tigr.antware.shared.command.events.SubmitEvent;
import org.tigr.antware.shared.command.events.SuspendEvent;
import org.tigr.htc.cmd.CmdState;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.CommandHelper;
import org.tigr.htc.cmd.CommandSetRoot;
import org.tigr.htc.cmd.CommandType;
import org.tigr.htc.cmd.TaskGroup;
import org.tigr.htc.common.HTCConfig;

/**
 * The <b>RequestMonitor</b> class handles the communications with 
 * the HTCServer. It is responsible for submitting command and then
 * registering a proxy listener that can receive events from the
 * server. The proxy listener then routes the events to the 
 * correct request objects.
 *
 */

public class RequestMonitor implements IProxyServer {
    static Logger log = Logger.getLogger(RequestMonitor.class);
    
    /**
     * <code>clientPort</code> starting port for client
     *
     */
    static int clientPort = 9001;
    
    /**
     * <code>MAX_PORT</code> try clientPort to MAX_PORT 
     *
     */
    static final int MAX_PORT = 11090;
    
    /**
     * <code>ENV_CMD</code> command to get local environment
     *
     */
    static final String ENV_CMD = "/usr/bin/env";
    
    /**
     * <code>env</code> save environment here
     *
     */
    static HashMap <String, String> env;
    
    // self, singleton
    static private RequestMonitor mon;
    
    // client xml-rpc object to send procedure calls to server
    static private XmlRpcClient client; 
    
    // webserver to receive events back from server
    static private WebServer webserver;
    
    // local hostname needed to send to the server
    static protected String hostname;
    
    // list of jobs to monitor
    static private Hashtable requests;
    
    // proxy listener
    private CmdEventProxy proxy;
    /**
     * The variable <code>local</code> holds the local submission status of the server  
     */
    private static boolean local = false;    

    
    /**
     * The variable <code>passEnvironment</code> hold the configuration value for this configuration
     */
    private boolean passEnvironment = true;    
    /**
     * inner class <code>CmdEventProxy</code> handles events send from
     * server. It finds the right request object that the event if meant for
     * and forwards it on. There is one method for each event type.
     *
     */
    public class CmdEventProxy {
        
        public int commandStarted(String id, String message, Date date,
				String mac) {
            log.debug("Got callback for commandStarted ");
            Long longId = Long.valueOf(id);
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.get(longId);
                StartEvent event =
                    new StartEvent(req, longId.longValue(), date, message, mac);
                req.cmd.fireStartEvent(event);
                return 0;
            } else {
                log.warn("Start event for command(" + id + ") that doesn't exist");
                return -1;
            }
            
        }
        
        public int commandFinished(
                String id, String message, Date date, int returnValue) {
            log.debug("Got callback for commandFinished ");
            Long longId = Long.valueOf(id);
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.remove(longId);
                FinishEvent event =
                    new FinishEvent(
                            req,
                            longId.longValue(),
                            date,
                            message,
                            returnValue);
                req.cmd.fireFinishEvent(event);
                return 0;
            } else {
                log.warn("Finish event for Command(" + id + ") that doesn't exist");
                return -1;
            }
        }
        
        public int commandFailed(String id, String message, Date date,
				String exception) {
            log.debug("Got callback for commandFailed.");
            Long longId = Long.valueOf(id);
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.remove(longId);
                FailureEvent event =
                    new FailureEvent(
                            req,
                            longId.longValue(),
                            date,
                            message,
                            new Exception(exception));
                req.cmd.fireFailureEvent(event);
                return 0;
            } else {
                log.warn("Failed event for Command(" + id + ") that doesn't exist");
                return -1;
            }
        }
        
        public int commandSubmitted(String id, String message, Date date,
				String machine, String user, String p_log) {
            log.debug("Got callback for commandSubmitted.");
            Long longId = Long.valueOf(id);
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.get(longId);
                SubmitEvent event =
                    new SubmitEvent(
                            req,
                            longId.longValue(),
                            date,
                            message,
                            machine,
                            user,
                            p_log);
                req.cmd.fireSubmitEvent(event);
                return 0;
            } else {
                log.warn("Submit event for Command(" + id + ") that doesn't exist");
                return -1;
            }
        }
        
        public int commandInterrupted(
                String id, String message, Date date, String interrupt) {
            log.debug("Got callback for commandInterrupted.");
            Long longId = Long.valueOf(id);
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.remove(longId);
                InterruptEvent event =
                    new InterruptEvent(
                            req,
                            longId.longValue(),
                            date,
                            message,
                            CommandInterrupt.getCommandInterrupt(interrupt));
                req.cmd.fireInterruptEvent(event);
                return 0;
            } else {
                log.warn("Interrupted event for Command(" + id + ") that doesn't exist");
                return -1;
            }
        }
        
        public int commandSuspended(String id, String message, Date date,
				String suspend) {
        	int suspended = 0;
            log.debug("Got callback for commandSuspended.");
            Long longId = Long.valueOf(id);
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.get(longId);
                SuspendEvent event =
                    new SuspendEvent(
                            req,
                            longId.longValue(),
                            date,
                            message,
                            CommandSuspend.getCommandSuspend(suspend));
                req.cmd.fireSuspendEvent(event);
            } else {
                log.warn("Suspended event for Command(" + id + ") that doesn't exist");
                suspended = -1;
            }
            return suspended;
        }
        
        public int commandResumed(String id, String message, Date date, String mac) {
            Long longId = Long.valueOf(id);
            int resumed = 0;
            
            if (requests.containsKey(longId)) {
                HTCRequest req = (HTCRequest) requests.get(longId);
                ResumeEvent event = new ResumeEvent(req, longId.longValue(), date, message, mac);
                req.cmd.fireResumeEvent(event);
            } else {
                resumed = -1;
            }
            return resumed;
        }
    }
    
    /**
     * Creates a new <code>RequestMonitor</code> instance. The RequestMonitor
     * is a singleton so the constructor is private.  
     *
     * @param p_uri a <code>String</code> value
     * @exception java.net.MalformedURLException if create XML-RPC server
     * @exception java.net.UnknownHostException if can't get local hostname
     * @exception IOException if a network problem occurs
     */
    private RequestMonitor(String p_uri) 
    throws java.net.MalformedURLException, 
    java.net.UnknownHostException, 
    IOException {
        log.debug("Make a new Request monitor for server " + p_uri);
        
        hostname = InetAddress.getLocalHost().getHostAddress();
        requests = new Hashtable();
        client = new XmlRpcClient(p_uri);
        proxy = new CmdEventProxy();

        // Get the configuration for passEnvironment from the config file if specified
        // Else set it to true
        if (HTCConfig.getProperty("grid.passenv") != null) {
        	passEnvironment = Boolean.parseBoolean(HTCConfig.getProperty("grid.passenv"));
        	log.info("The directive for passing environment by default to grid jobs is: " + passEnvironment);
        }
        
        if (passEnvironment) {
			// Grab local the environment in case the user wants the local environment to be passed along
			log.debug("Grab the local environment");
			Map<String, String> envMap = System.getenv();
			// Iterate over the environment and add to the env vector. While adding stip some of the
			// environment variables known to cause problems such as LS_COLORS, HOST, HOSTTYPE, HOSTNAME
			env = new HashMap<String, String>();
			if (log.isDebugEnabled()) {
				log.debug("----- start environment ----");
			}
			for (String envVar : envMap.keySet()) {

				if (!(envVar.equals("LS_COLORS") || envVar.equals("HOST")
						|| envVar.equals("HOSTTYPE") || envVar
						.equals("HOSTNAME"))) {
					// As the variable is not one of the ones that is skipped add to the vector
					env.put(envVar, envMap.get(envVar));

					if (log.isDebugEnabled()) {
						log.debug(envVar + " : " + envMap.get(envVar));
					}
				}
			}
			if (log.isDebugEnabled()) {
				log.debug("----- end environment ----");
			}
		}
        
        // Create XML-RPC server on a unused port
        // start at a default port, if that is used try the next port
        log.debug("Starting client callback server");
        while (true) {
            try {
                log.debug("Try port " + clientPort);
                webserver = new WebServer(clientPort);
                webserver.addHandler("Request", proxy);
                break;
            } catch (Exception e) {
                log.warn("Port taken " + clientPort + " try next port ");
                if (++clientPort > MAX_PORT) {
                    log.fatal(" no more ports to try ");
                    throw new IOException("Can't find a free port on the client machine.");
                }
            }
        }
        
        // might move to start server after first submit
        log.debug("Calling start on the webserver.");
        webserver.start();
        
        // Sleep for a little bit to give the webserver thread time to start as occasionally
        // the service sends events before the webserver is listening. This is not guaranteed
        // to work, but stands a better chance
        try {
            Thread.sleep(5000);
        } catch (Exception e) {
        	// ignored
        }
    }
    
    
    /**
     * <code>getInstance</code> get or create the one and only
     * a RequestMonitor instance.
     *
     * @param p_uri a <code>String</code> value
     * @return a <code>RequestMonitor</code> value
     * @exception java.net.MalformedURLException if an error occurs
     * @exception java.net.UnknownHostException if an error occurs
     * @exception IOException if an error occurs
     */
    public synchronized static RequestMonitor getInstance(String p_uri)
			throws java.net.MalformedURLException, UnknownHostException, IOException {
        if (mon == null) {
            mon = new RequestMonitor(p_uri);
        }
        return mon;
    }
    
    // 
    // Start of IProxyServer methods
    //
    
    /**
     * <code>submit</code> submit the request to the grid.  If the request is valid and
     * is accepted by the cluster a unique positive id will be returned.  If there is 
     * error or the request is rejected the id returned will be less then 1.
     *
     * @return a <code>long</code> a unique id assigned to the request
     * @exception IllegalAccessException if a request has already been submitted
     * @exception IOException if a network problem occurs
     */
    public long submit(HTCRequest request) throws IllegalAccessException,
			IOException {
        String result = "";
        
        try {
            log.debug("In submit, calling createCmdID");
            
            // Vector for XML-RPC params
            Vector params = new Vector();
            
            // get a new id and add command to monitor
            String idStr = (String) client.execute("HTCServer.createCmdID", params);
            log.debug("Created an id of " + idStr);
            Long id = Long.valueOf(idStr);
            
            // Check if id is valid. A Valid id should be greater than 0.
            if (id.longValue() <= 0) {
            	// If id is not valid, throw an exception
            	throw new IOException("Failed to get a valid request id.");
            }
            
            // Set the ID to command object
            request.cmd.setID(id.longValue());
            //Set submit URL
            request.cmd.setSubmitURL("http://" + hostname + ":" + clientPort + "/RPC2");
            
            log.debug("The Command Q is: " + 
                    request.cmd.getConfig().getCmdClass());
            
            // Add the current request to the list of requests that are launched
            // so they can be monitored
            if (! requests.containsKey(id)) {
                log.debug("Add request to hash.");
                requests.put(id, (Object) request);
            }
            CommandSetRoot csr = new CommandSetRoot();
            
            log.debug("Create XML-RPC request");
            log.debug("Create CommandSet root and add as the first arg");
            csr.setCommand(request.cmd);
            params.add(csr.toString());
            
            log.debug("Add callback URL as second argument.");
            params.add("http://" + hostname + ":" + clientPort + "/RPC2");
            
            // Should we send the environment? 
            if (request.isGetEnv()) {
                log.debug("getenv is set");
                if (request.getEnv() != null) {
                    params.add(request.getEnv());
                } else {
                    params.add(RequestMonitor.env);
                }
            }


            
            // Set the environment to the Command that is being executed based on the user settings or global
            // configuration. If the user has specified that the environment should be set in the request then
            // honor that, else use the global configuration file settings.
            if (request.isGetEnv()) {
            	if (log.isDebugEnabled()) {
            		log.debug("Get environment property set to true on command. Will pass environemt.");
            	}
            	
            	Map <String, String> envMap;
            	// Set the user specified environment
                if (request.getEnv() != null) {
                	if (log.isDebugEnabled()) {
    					log.debug("Setting user specified environment");
    				}
                	envMap = (Map<String, String>) request.getEnv();
                } else {
                	// Set the default environment that was grabbed by the 
                	// SGEMockServer during the initialization as the user has not specified
                	// the environment
                	if (log.isDebugEnabled()) {
    					log.debug("Setting local environment");
    				}
                	envMap = env;
                }
                
		    	for (String envVar : envMap.keySet()) {
		    		params.add(envVar + "=\"" + envMap.get(envVar) + "\"");
				}
            }
            else if (passEnvironment) {
            	// As the global configuration specifies passing an environment pass the local
            	// environment
            	if (log.isDebugEnabled()) {
    				log.debug("Setting local environment");
    			}
                
		    	for (String envVar : env.keySet()) {
		    		params.add(envVar + "=\"" + env.get(envVar) + "\"");
				}
            }
            
            // Make XML-RPC call, id string is returned, XML-RPC doesn't support longs
            result = (String) client.execute("HTCServer.submit", params);
            log.debug("Request Monitor submitted Command and result was " + result);
            Long.parseLong(result);
            
            return id.longValue();
        } catch (IOException e) {
            log.warn("I/O error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("XmlRpcException in RequestMonitor.submit  " + e.getMessage());
            throw new IOException("Server side error: " + e.getMessage());
        } catch (NumberFormatException e) {
            log.warn(
                    "NumberFormatException in RequestMonitor.submit  "
                    + result
                    + e.getMessage());
            throw new IOException("Submission rejected with message: " + result);
        } finally {
            // AAM: What happens if a submit fails? Seems like there is no mechanism
            // for the request to be removed, or do we always get a call back on submit
            // failure which will remove the request from the hash
            // Remove the request if there was any exception
        }
    }
    
    /**
     * <code>getCommand</code> ask the server for the command that is associated with
     * the id.
     * 
     * @param id
     *            a <code>long</code> id for an existing command
     * @return a <code>Command</code> the command object
     * @exception IOException
     *                if an network problem occurs
     */
    public Command getCommand(long id, HTCRequest request) throws IOException {
        try {
        	
        	log.debug("Getting Command object from server for id " + id );
        	
            Vector params = new Vector();
            params.add(Long.toString(id));
            String cmdStr = (String) client.execute("HTCServer.getCommand", params);
            log.info(cmdStr);
            // Convert XML-RPC string to stream
            StringReader fr = new StringReader(cmdStr);
            
            // Unmarshal XML into objects
            Command cmd = (Command) CommandHelper.getInstance().unmarshal((Reader) fr);
            
            if(cmd.getState() != CmdState.FINISHED) {
                Long CID = new Long(cmd.getID());
                if(!requests.containsKey(CID)) {
                    log.debug("Add request to hash.");
                    requests.put(CID, (Object) request);
                }
            }
            
            return cmd;
        } catch (IOException e) {
            log.warn("I/O error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("XmlRpcException in getCommand() " + e.getMessage());
            throw new IOException(e.getMessage());
        } catch (Exception e) {
            log.warn("Exception in getCommand() " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
    }
    
    public boolean isNFS(String p_path) throws IOException {
        try {
            Vector params = new Vector();
            params.add(p_path);
            Boolean b = (Boolean) client.execute("HTCServer.isNFS", params);
            return b.booleanValue();
        } catch (IOException e) {
            log.warn("isNFS I/O error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("XmlRpcException in isNFS " + e.getMessage());
            return false;
        }
    }
    
    // call server to get status on a command
    public Map getStatus(HTCRequest p_request) {
        try {
            Vector params = new Vector();
            params.add(Long.toString(p_request.cmd.getID()));
            Map b = (Map) client.execute("HTCServer.status", params);
            return b;
        } catch (Exception e) {
            log.debug("Request monitor error ", e);
            return null;
        }
    }
    
    public String getMonitorHost(HTCRequest p_request) {
        try {
            Vector params = new Vector();
            params.add(Long.toString(p_request.cmd.getID()));
            String b = (String) client.execute("HTCServer.monitorHost", params);
            return b;
        } catch (Exception e) {
            log.debug("Request monitor error ", e);
            return null;
        } 
    }
    
	public TaskGroup getTasks(HTCRequest request) throws IOException {
	    if (! request.getCommandObj().getType().equals(CommandType.MW)) {
            throw new IOException("Attempt to request tasks on non-array type job.");
        }
	    
	    String b = "";
        try {
            Vector params = new Vector();
            params.add(Long.toString(request.cmd.getID()));
            b = (String) client.execute("HTCServer.getTasks", params);
        } catch (Exception e) {
            log.warn("Exception in getTasks  " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
        
        // Check to see if there was an error from the server side which
        // typically happens because the command cannot be found then throw exception
        if (b.equals("Command not found")) {
            throw new IOException(b);
        }
        
        try {
            TaskGroup tg = (TaskGroup) CommandHelper.getInstance().unmarshal(
                    new StringReader(b));
            return tg;
        } catch (Exception e) {
            log.warn("Exception in getTasks() " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
	}
	
    /**
     * <code>resume</code> send a resume command to a request that has been
     * held/suspended by the user. This will cause the request to be put back in a queue
     * to be scheduled. This method doesn't wait for the command to stop before it
     * returns. Delegated to proxy.
     * 
     * @exception IllegalAccessException
     *                if set is attempted after submission
     */
    public void resume(HTCRequest request) throws IOException {
        try {
            Vector params = new Vector();
            params.add(Long.toString(request.cmd.getID()));
            Boolean b = (Boolean) client.execute("HTCServer.resume", params);
            return;
            
        } catch (IOException e) {
            log.warn("I/O error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("XmlRpcException in resume()  " + e.getMessage());
            throw new IOException(e.getMessage());
        } catch (Exception e) {
            log.warn("Exception in resume: " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
    }
    
    /**
     * <code>stop</code> send a stop to the request. This method doesn't wait for the
     * command to stop before it returns.
     * 
     * @exception IOException
     *                if the request hasn't been submitted
     */
    public void stop(HTCRequest request) throws IOException {
        try {
            log.debug("In stop on " + request.cmd.getID());
            Vector params = new Vector();
            params.add(Long.toString(request.cmd.getID()));
            Boolean b = (Boolean) client.execute("HTCServer.stop", params);
            log.debug("stop returned " + b);
            return;
        } catch (IOException e) {
            log.warn("I/O error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("XmlRpcException in stop() " + e.getMessage());
            throw new IOException(e.getMessage());
        } catch (Exception e) {
            log.warn("Exception in stop() " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
    }
    
    /**
     * <code>hold</code> send a hold/suspend command to a request. Delegated to proxy.
     * 
     * @exception IOException
     *                if set is attempted after submission
     */
    public void hold(HTCRequest request) throws IOException {
        try {
            Vector params = new Vector();
            params.add(Long.toString(request.cmd.getID()));
            Boolean b = (Boolean) client.execute("HTCServer.hold", params);
            return;
            
        } catch (IOException e) {
            log.warn("RequestMonitor.hold io error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("RequestMonitor.xmlrpcexception in hold() " + e.getMessage());
            throw new IOException(e.getMessage());
        } catch (Exception e) {
            log.warn("Exception in hold() " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
    }
    
    /**
     * <code>disconnect</code> client no longer wants event updates from server.
     * delegate to proxy server.
     * 
     * @exception IllegalAccessException
     *                if set is attempted after submission
     */
    public void disconnect(HTCRequest request) throws IOException {
        try {
            Vector params = new Vector();
            params.add(Long.toString(request.cmd.getID()));
            String b = (String) client.execute("HTCServer.disconnect", params);
            return;
        } catch (IOException e) {
            log.warn("I/O error with XML-RPC " + e.getMessage());
            throw e;
        } catch (XmlRpcException e) {
            log.warn("XmlRpcException in disconnect() " + e.getMessage());
            throw new IOException(e.getMessage());
        } catch (Exception e) {
            log.warn("Exception in disconnect() " + e.getMessage(), e);
            throw new IOException(e.getMessage());
        }
    }

    /**
     * The method <code>isLocal</code> returns the status of the proxy server. For local submissions this 
     * value is true and for remote submissions the value is false.
     *  
     * @return boolean true to indicate the server is performing local submits
     */
	public boolean isLocal() {
		return local;
	}

    /**
     * The method <code>isPassEnvironment</code> provides the configuration for the proxy server
     * to pass the local environment to the grid jobs. This is read from a global configuration file.
     * 
     * @return a <code>boolean</code> of true if the environment is explicitly passed to the grid jobs
     * when the environment is not set by the user.
     */
	public boolean isPassEnvironment() {
		return passEnvironment;
	}
}
